package cn.yunyichina.scheduler.jobs.service.pay.impl;

import cn.yunyichina.provider.framework.dubbo.service.PayService;
import cn.yunyichina.provider.framework.kafka.KafkaTopic;
import cn.yunyichina.provider.iface.entity.common.Response;
import cn.yunyichina.scheduler.jobs.constant.MethodRouter;
import cn.yunyichina.scheduler.jobs.service.pay.CheckingAccountsService;
import cn.yunyichina.utils.commons.DateUtils;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * 对账定时调度
 * Created by JoeZ on 2016/11/17.
 */
@Service
public class CheckingAccountsServiceImpl implements CheckingAccountsService {

    private static final Logger logger = LoggerFactory.getLogger(CheckingAccountsServiceImpl.class);

    @Resource
    private KafkaTemplate<Integer, String> kafkaTemplate;

    private final String TOPIC_NAME = "CHECK_ACCOUNT_START";

    public static SimpleDateFormat YYYYMMDDHHMISS14 = new SimpleDateFormat("yyyyMMddHHmmss");

    @Override
    public void doExecute() {
        produceMessage(TOPIC_NAME , "START");
    }


    /**
     * 产出支付通知消息到消息队列(业务方监听消息完成支付闭环)
     *
     * @param topic
     * @param message
     */
    private void produceMessage(String topic, String message) {
        logger.info("对账定时器执行开始");
        if (StringUtils.isEmpty(topic)) return;
        logger.info("CheckingAccountsServiceImpl.produceMessage.topic -> {}", topic);

        String kafkaTopic = KafkaTopic.valueOf(topic).getTopic();
        logger.info("CheckingAccountsServiceImpl.produceMessage.kafkaTopic -> {}", kafkaTopic);

        ListenableFuture<SendResult<Integer, String>> listenableFuture = kafkaTemplate.send(kafkaTopic, message);

        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            public void onSuccess(SendResult<Integer, String> sendResult) {
                logger.info("对账定时器推送成功！");
                logger.info("CheckingAccountsServiceImpl.produceMessage.onSuccess.getProducerRecord -> {}", sendResult.getProducerRecord().toString());
                logger.info("CheckingAccountsServiceImpl.produceMessage.onSuccess.getRecordMetadata -> {}", sendResult.getRecordMetadata().toString());
            }

            public void onFailure(Throwable throwable) {
                logger.error("CheckingAccountsServiceImpl.produceMessage.onFailure", throwable);
                Map<String, String> map = new HashMap<>();
                map.put("orgId","ALL");
                map.put("eventCode","CK_TIMER_ERROR");
                map.put("hospitalName","所有医院");
                map.put("date", DateUtils.getDateStr(DateUtils.yesterdayDate()));
                map.put("batch",YYYYMMDDHHMISS14.format(new Date()));
                map.put("level","3");
                map.put("problemDescription","定时器推送消息队列消息失败！");
                PayService.callHumpFromInternal(MethodRouter.SEND_FAILSMS_PAY, JSON.toJSONString(map));

                Map<String, String> logMap = new HashMap<>();
                logMap.put("orgId","ALL");
                logMap.put("eventCode","CK_TIMER_ERROR");
                logMap.put("errorMessage","定时器推送消息队列消息失败！");
                PayService.callHumpFromInternal(MethodRouter.INPUT_FAILLOG_PAY, JSON.toJSONString(map));
            }
        });

        logger.info("对账定时器执行结束");
    }
}
